package com.jaymin.flink.app;

import cn.hutool.json.JSONUtil;
import com.jaymin.flink.domain.Access;
import com.jaymin.flink.sink.RedisExampleMapper;
import com.jaymin.flink.sink.RedisExampleMapperV2;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import java.util.Objects;

/**
 * 按操作系统的维度进行新老用户的分析
 *
 * @author jaymin
 * @since 2024/4/5 23:26
 */
public class OsUserCntAppV1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<Access> streamOperator = executionEnvironment.readTextFile("data/assess_info.json")
                .map(new MapFunction<String, Access>() {
                    @Override
                    public Access map(String value) throws Exception {
                        Access access = JSONUtil.toBean(value, Access.class);
                        return access;
                    }
                })
                .filter(x -> x != null)
                .filter(x -> Objects.equals(x.getEvent(), "startup"));
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> result = streamOperator.map(new MapFunction<Access, Tuple3<String, Integer, Integer>>() {
                    @Override
                    public Tuple3<String, Integer, Integer> map(Access value) throws Exception {
                        return Tuple3.of(value.getOs(), value.getNu(), 1);
                    }
                }).keyBy(new KeySelector<Tuple3<String, Integer, Integer>, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> getKey(Tuple3<String, Integer, Integer> value) throws Exception {
                        return Tuple2.of(value.f0, value.f1);
                    }
                })
                .sum(2);
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build();
        result.addSink(new RedisSink<Tuple3<String, Integer,Integer>>(conf, new RedisExampleMapperV2()));
        executionEnvironment.execute("OsUserCntAppV1");
    }
}
